package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"
	"bytes"
	"fmt"
	"log"
	"math/rand"
	"os"
	"sync"
	"sync/atomic"
	"time"

	//	"6.5840/labgob"
	"6.5840/labgob"
	"6.5840/labrpc"
)

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type LogEntry struct {
	Term 			int		// leader’s term
	PrevLogIndex	int		// index of log entry immediately preceding	new ones
	PrevLogTerm		int		// term of prevLogIndex entry
	Command			interface{}
}

type settableIndex struct {
	mu				sync.Mutex
	idx				int
	setTime			int64
}

type atomInt struct {
	sync.Mutex
	cond			*sync.Cond
	value			int
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu					sync.RWMutex        // Lock to protect term, status, voteFor, logs
											// lock order: rf.commitIndex.Lock() -> rf.mu.Lock()
	cond				*sync.Cond			// Conditional variable on status
	peers				[]*labrpc.ClientEnd // RPC end points of all peers
	persister			*Persister          // Object to hold this peer's persisted state
	me					int                 // this peer's index into peers[]
	dead				int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	term				int
	status	  			int32				// 0: follower 1: candidate 2:leader
	voteFor				int
	logs	  			[]LogEntry

	commitIndex			atomInt				// index of highest log entry known to be committed
											// (initialized to -1, increases monotonically)
											// lock order: rf.commitIndex.Lock() -> rf.mu.Lock()
	lastApplied			int					// index of highest log entry applied to state machine
											// (initialized to -1, increases monotonically)
											// no need to lock because only `apply` function access it
	append_flag_cnt		atomic.Int64		// increase to indicate receive an appendlog
	requestVoteTerm		atomic.Int32		// current term which RequestVote for, -1 represents no RequestVote
	appendEntryTerm		atomic.Int32		// current term which HeartBeat or AppendEntry for, -1 represents no HeartBeat
	rountineCnt			atomic.Int32		// sum of routine number, only for debug

	// Volatile state on leaders
	nextIndex			[]settableIndex		// for each server, index of the next log entry to send
											// to that server (initialized to leader last log index + 1)
	matchIndex			[]settableIndex		// for each server, index of highest log entry known to be 
											// replicated on server (initialized to 0, increases monotonically)

	// Logger
	logger *log.Logger
}


func (rf *Raft) initLogger() {
    // Initialize the logger
	filename := fmt.Sprintf("server%d.log", rf.me)
    file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0666)
    if err != nil {
        log.Fatalf("Failed to open log file: %v", err)
    }
    // Set log flags to include the date, time, and file short name with line number
    rf.logger = log.New(file, "INFO: ", log.Ldate|log.Ltime|log.Lmicroseconds)
}

func (rf *Raft) logMessage(message string) {
    rf.logger.Println(message)
}

// args[0]: line
func (rf *Raft) Lock() {
	rf.mu.Lock()
	// _, _, line, ok := runtime.Caller(1)
	// if ok {
	// 	rf.logMessage(fmt.Sprintf("Lock in line%d", line))
	// } else {
	// 	rf.logMessage("lock in unknown line")
	// }
}

func (rf *Raft) Unlock() {
	rf.mu.Unlock()
	// _, _, line, ok := runtime.Caller(1)
	// if ok {
	// 	rf.logMessage(fmt.Sprintf("Unlock in line%d", line))
	// } else {
	// 	rf.logMessage("Unlock in unknown line")
	// }
}

func (rf *Raft) RLock() {
	rf.mu.RLock()
	// _, _, line, ok := runtime.Caller(1)
	// if ok {
	// 	rf.logMessage(fmt.Sprintf("RLock in line%d", line))
	// } else {
	// 	rf.logMessage("RLock in unknown line")
	// }
}

func (rf *Raft) RUnlock() {
	rf.mu.RUnlock()
	// _, _, line, ok := runtime.Caller(1)
	// if ok {
	// 	rf.logMessage(fmt.Sprintf("RUnlock in line%d", line))
	// } else {
	// 	rf.logMessage("RUnlock in unknown line")
	// }
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.RLock()
	defer rf.RUnlock()
	// Your code here (2A).
	return rf.term, rf.status==2
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.term)
	e.Encode(rf.voteFor)
	e.Encode(rf.logs)
	e.Encode(rf.status)
	raftstate := w.Bytes()
	rf.persister.Save(raftstate, nil)
}


// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var term int
	var voteFor int
	var logs []LogEntry
	var status int32
	if d.Decode(&term) != nil ||
	   d.Decode(&voteFor) != nil || d.Decode(&logs) != nil || d.Decode(&status) != nil {
		rf.logMessage("readPersist decode error!")
		os.Exit(1)
	} else {
	  	rf.term = term
	  	rf.voteFor = voteFor
		rf.logs = logs
		rf.status = status
	}
}


// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}


// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term			int
	CandidateId		int
	LastLogIndex	int
	LastLogTerm		int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term			int
	VoteGranted		bool
}

type AppendEntriesArgs struct {
	Term			int
	LeaderId		int		// so follower can redirect clients
	PrevLogIndex	int		// index of log entry immediately preceding new ones
	PrevLogTerm		int		// term of prevLogIndex entry
	Entries[]		LogEntry
	LeaderCommit	int		// leader’s `commitIndex`
}

type AppendEntriesReply struct {
	Term		int		// currentTerm, for leader to update itself
	XTerm		int		// term in the conflicting entry (if any)
	XIndex		int		// index of first entry with that term (if any)
	XLen		int		// log length
	SetTime		int64	// AppendEntry rpc return time, used to label which reply is outdated
	TermError	bool	// true if follwer's term > leader's term
	Success		bool	// true if append successfully(entry != nil) or log check only(hearbeat) successfully
}

func MyMinInt(a int, b int) int{
	if a > b {
		return b
	}
	return a
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.rountineCnt.Add(1)
	defer rf.rountineCnt.Add(-1)
	// Your code here (2A, 2B).
	do_persist := false
	rf.Lock()

	if rf.term < args.Term {
		rf.term = args.Term
		rf.voteFor = -1
		rf.status = 0
		rf.cond.Broadcast()
		rf.requestVoteTerm.Store(-1)
		rf.appendEntryTerm.Store(-1)
		rf.logMessage(fmt.Sprintf("during handling RequestVote from server%d, setting term to %d, i become follower", args.CandidateId, rf.term))
		do_persist = true
	}
	reply.Term = rf.term

	// if rf.term == args.Term, then grant vote if rf.voteFor is -1(follower), 
	// if rf.term < args.Term, then rf.voteFor must be -1 because reset it in previous code block
	if rf.voteFor != -1 || rf.term > args.Term {
		if rf.voteFor != -1 {
			rf.logMessage(fmt.Sprintf("during handling RequestVote from server%d, rf.voteFor is not -1, VoteGranted false\n", args.CandidateId))
		} else {
			rf.logMessage(fmt.Sprintf("during handling RequestVote from server%d, rf.term: %d, args.Term: %d, rf.term > args.Term, VoteGranted false\n", args.CandidateId, rf.term, args.Term))
		}
		rf.Unlock()
		reply.VoteGranted = false
	} else if len(rf.logs) == 0 || rf.logs[len(rf.logs) - 1].Term < args.LastLogTerm || (rf.logs[len(rf.logs) - 1].Term == args.LastLogTerm && len(rf.logs) <= args.LastLogIndex + 1) {
		rf.voteFor = args.CandidateId
		rf.persist()
		rf.Unlock()
		reply.VoteGranted = true
		rf.append_flag_cnt.Add(1) // a valid requestVote rpc only if its logs is also up to date
	} else {
		if do_persist {
			rf.persist()
		}
		rf.logMessage(fmt.Sprintf("during handling RequestVote from server%d, VoteGranted false, len(rf.logs): %d, last log's Term: %d, args.LastLogIndex: %d, args.LastLogTerm: %d\n", args.CandidateId, len(rf.logs), rf.logs[len(rf.logs) - 1].Term, args.LastLogIndex, args.LastLogTerm))
		rf.Unlock()
		reply.VoteGranted = false
	}
	rf.logMessage(fmt.Sprintf("receive RequestVote from server%d, I vote %t for it", args.CandidateId, reply.VoteGranted))	
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.rountineCnt.Add(1)
	defer rf.rountineCnt.Add(-1)

	do_persist := false
	rf.Lock()
	if rf.term <= args.Term {
		rf.append_flag_cnt.Add(1)
	}

	if rf.term < args.Term || (rf.status == 1 && rf.term == args.Term) {
		rf.term = args.Term
		rf.voteFor = -1
		rf.status = 0
		do_persist = true
		rf.cond.Broadcast()
		rf.requestVoteTerm.Store(-1)
		rf.appendEntryTerm.Store(-1)
		rf.logMessage(fmt.Sprintf("during handling AppendEntries from server%d, setting term to %d, i become follower", args.LeaderId, rf.term))
	}

	reply.Term = rf.term
	if rf.term > args.Term {
		reply.TermError = true
		reply.Success = false
	} else if len(rf.logs) <= args.PrevLogIndex {
		reply.TermError = false
		reply.Success = false
		reply.XIndex = -1
		reply.XTerm = -1
		reply.XLen = len(rf.logs)
	} else if args.PrevLogIndex >= 0 && rf.logs[args.PrevLogIndex].Term != args.PrevLogTerm {
		reply.TermError = false
		reply.Success = false
		reply.XTerm = rf.logs[args.PrevLogIndex].Term
		reply.XLen = len(rf.logs)
		// binary search
		st, ed := 0, len(rf.logs) - 1
		for st < ed {
			mid := (st + ed) / 2
			if rf.logs[mid].Term < reply.XTerm {
				st = mid + 1
			} else {
				ed = mid
			}
		}
		reply.XIndex = ed
	} else {
		reply.TermError = false
		if args.Entries != nil {
			// Debug only
			if args.Entries[0].PrevLogIndex != args.PrevLogIndex {
				rf.logMessage("exit because log index not match!!")
				os.Exit(1)
			}

			if len(rf.logs) > args.PrevLogIndex + 1 && rf.logs[args.PrevLogIndex + 1].Term != args.Entries[0].Term {
				rf.logs = rf.logs[:args.PrevLogIndex + 1]
				do_persist = true
			}

			if len(rf.logs) <= args.PrevLogIndex + 1 {
				rf.logs = append(rf.logs, args.Entries[0])
				do_persist = true
			}
		}
		reply.Success = true
	}
	pre_index := len(rf.logs) - 1
	if do_persist {
		rf.persist()
	}
	rf.Unlock()
	reply.SetTime = time.Now().UnixNano()

	rf.logMessage(fmt.Sprintf("receive AppendEntries from server%d", args.LeaderId))

	if reply.Success {
		rf.commitIndex.Lock()
		if rf.commitIndex.value < args.LeaderCommit {
			rf.commitIndex.value = MyMinInt(args.LeaderCommit, pre_index)
			rf.logMessage(fmt.Sprintf("set rf.commitIndex to %d", rf.commitIndex.value))
			rf.commitIndex.cond.Signal()
		}
		rf.commitIndex.Unlock()
	}
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1

	// Your code here (2B).
	rf.Lock()

	term = rf.term
	if rf.status != 2 {
		rf.Unlock()
		return index, term, false
	}
	pre_index := len(rf.logs) - 1
	index = pre_index + 1
	pre_log_term := 0
	if pre_index >= 0 {
		pre_log_term = rf.logs[pre_index].Term
	}
	new_entry := LogEntry{Term: term, PrevLogIndex: pre_index, PrevLogTerm: pre_log_term, Command: command}
	rf.logs = append(rf.logs, new_entry)
	rf.persist()
	rf.Unlock()

	// test system expect index start from 1, so return (index + 1)
	return index + 1, term, true
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) TransToLeader(term int, lastLogIndex int, lastLogTerm int) {
	rf.rountineCnt.Add(1)
	defer rf.rountineCnt.Add(-1)
	var ticket atomic.Int32
	ticket.Store(1)

	for i := 0; i < len(rf.peers) && int(rf.requestVoteTerm.Load()) == term && !rf.killed(); i++ {
		if i != rf.me {
			go func(i int) {
				rf.rountineCnt.Add(1)
				defer rf.rountineCnt.Add(-1)
				ok := false
				retryCnt := 0
				for !ok && int(rf.requestVoteTerm.Load()) == term && !rf.killed() {
					args := &RequestVoteArgs{term, rf.me, lastLogIndex, lastLogTerm}
					reply := &RequestVoteReply{}
					start := time.Now() // Record the start time
					ok = rf.sendRequestVote(i, args, reply)
					elapsed := time.Since(start)
					rf.logMessage(fmt.Sprintf("sendRequestVote to server %d return %v, term: %d, reply.VoteGranted %v, reply.Term %v, time cost: %s", i, ok, term, reply.VoteGranted, reply.Term, elapsed))
					if ok {
						if reply.VoteGranted {
							ticket.Add(1)
						} else if term < reply.Term {
							rf.Lock()
							if rf.term < reply.Term {
								rf.term = reply.Term
								rf.voteFor = -1
								rf.status = 0
								rf.persist()
								rf.cond.Broadcast()
								rf.requestVoteTerm.Store(-1)
								rf.appendEntryTerm.Store(-1)
								rf.logMessage(fmt.Sprintf("during handling RequestVote response from server%d, setting term to %d", i, rf.term))
							}
							rf.Unlock()
						}
					} else {
						rf.logMessage(fmt.Sprintf("sendRequestVote to server %d failed, retryCnt: %d", i, retryCnt))
						retryCnt++
					}
				}
			}(i)
		}
	}

	for int(rf.requestVoteTerm.Load()) == term && !rf.killed() {
		if int(ticket.Load()) > len(rf.peers) / 2 {
			rf.Lock()
			if rf.term == term && rf.status == 1 {
				rf.status = 2
				rf.persist()
				currentTerm := rf.term
				rf.requestVoteTerm.Store(-1)
				rf.appendEntryTerm.Store(int32(currentTerm))
				log_len := len(rf.logs)
				rf.Unlock()
				for i:=0; i<len(rf.peers); i++ {
					rf.nextIndex[i].mu.Lock()
					rf.nextIndex[i].idx = log_len
					rf.nextIndex[i].setTime = time.Now().UnixNano()
					rf.nextIndex[i].mu.Unlock()
					rf.matchIndex[i].mu.Lock()
					rf.matchIndex[i].idx = -1
					rf.matchIndex[i].setTime = time.Now().UnixNano()
					rf.matchIndex[i].mu.Unlock()
				}
				rf.logMessage(fmt.Sprintf("I becomes leader, term %d", currentTerm))
				go rf.sendHeartBeatOrAppendEntries(currentTerm)
			} else {
				rf.Unlock()
			}
			break
		} else {
			time.Sleep(time.Duration(5) * time.Millisecond)
			// rf.logMessage(fmt.Sprintf("wake up with ticket: %d\n", ticket.Load()))
		}
	}

	rf.RLock()
	rf.logMessage(fmt.Sprintf("before TransToLeader return, status:%v, term: %v", rf.status, rf.term))
	rf.RUnlock()
}

func (rf *Raft) ticker() {
	rf.rountineCnt.Add(1)
	defer rf.rountineCnt.Add(-1)
	previous_cnt := int64(0)
	for !rf.killed() {

		// Your code here (2A)
		// Check if a leader election should be started.
		// pause for a random amount of time between 50 and 350
		// milliseconds.
		ms := 900 + (rand.Int63() % 300)
		time.Sleep(time.Duration(ms) * time.Millisecond)
		current_flag := rf.append_flag_cnt.Load()
		rf.logMessage(fmt.Sprintf("checking append_flag_cnt, current_flag: %d, previous_cnt: %d", current_flag, previous_cnt))
		rf.Lock()
		if rf.status == 1 || (rf.status == 0 && previous_cnt == current_flag) {
			rf.term++
			rf.requestVoteTerm.Store(int32(rf.term))
			rf.appendEntryTerm.Store(-1)
			term := rf.term
			lastLogIndex := len(rf.logs)-1
			lastLogTerm := 0
			if len(rf.logs) != 0 {
				lastLogTerm = rf.logs[len(rf.logs) - 1].Term
			} else {
				lastLogTerm = 0
			}
			rf.status = 1
			rf.voteFor = rf.me
			rf.persist()
			rf.Unlock()
			rf.logMessage(fmt.Sprintf("before transtoleader, term: %d, lastLogIndex: %d, lastLogTerm: %d", term, lastLogIndex, lastLogTerm))
			go rf.TransToLeader(term, lastLogIndex, lastLogTerm)
		} else if rf.status == 2 {
			rf.logMessage("before sleeping")
			rf.cond.Wait()
			rf.logMessage("after sleeping")
			rf.Unlock()
		} else {
			rf.Unlock()
		}
		previous_cnt = current_flag
	}
}

func (rf *Raft) sendHeartBeatOrAppendEntries(currentTerm int) {
	rf.rountineCnt.Add(1)
	defer rf.rountineCnt.Add(-1)

	matchIndexCnt := make(map[int]int)
	var matchIndexCntMutex sync.Mutex

	for i := 0; i < len(rf.peers) && int(rf.appendEntryTerm.Load()) == currentTerm && !rf.killed(); i++ {
		if i != rf.me {
			go func(i int) {
				rf.rountineCnt.Add(1)
				defer rf.rountineCnt.Add(-1)

				for int(rf.appendEntryTerm.Load()) == currentTerm && !rf.killed() {
					rf.nextIndex[i].mu.Lock()
					nextIndex := rf.nextIndex[i].idx
					rf.nextIndex[i].mu.Unlock()
					prevLogIndex := nextIndex - 1
					prevLogTerm := 0
					rf.RLock()
					if prevLogIndex >= 0 {
						prevLogTerm = rf.logs[prevLogIndex].Term
					}
					entries := []LogEntry(nil)
					if nextIndex < len(rf.logs) {
						entries = append(entries, rf.logs[nextIndex])
					}
					rf.RUnlock()

					ok := false
					
					rf.commitIndex.Lock()
					currentCommitIndex := rf.commitIndex.value
					rf.commitIndex.Unlock()

					args := &AppendEntriesArgs{Term: currentTerm, LeaderId: rf.me, PrevLogIndex: prevLogIndex, 
						PrevLogTerm: prevLogTerm, Entries: entries, LeaderCommit: currentCommitIndex}
					reply := &AppendEntriesReply{}
					start := time.Now() // Record the start time
					ok = rf.sendAppendEntries(i, args, reply)
					elapsed := time.Since(start)
					
					if ok {
						if entries == nil {
							rf.logMessage(fmt.Sprintf("send AppendEntry(heartbeat) to server %d, args: %v, reply: %v, time cost: %s", i, *args, *reply, elapsed))
						} else {
							rf.logMessage(fmt.Sprintf("send AppendEntry to server %d, args: %v, logIndex: %v, reply: %v, time cost: %s", i, *args, nextIndex, *reply, elapsed))
						}
						if reply.Success {
							if entries == nil { // heartbeat only
								rf.logMessage(fmt.Sprintf("send heartbeat to server %d successfully", i))
							} else { // append entry
								rf.logMessage(fmt.Sprintf("append entry to server %d successfully", i))
								// set nextIndex
								rf.nextIndex[i].mu.Lock()
								rf.logMessage(fmt.Sprintf("reply.SetTime: %s, rf.nextIndex[%d].setTime: %s, rf.nextIndex[%d].idx: %d", time.Unix(0, reply.SetTime), i, time.Unix(0, rf.nextIndex[i].setTime), i, rf.nextIndex[i].idx))
								if reply.SetTime > rf.nextIndex[i].setTime && rf.nextIndex[i].idx < nextIndex + 1 {
									rf.nextIndex[i].setTime = reply.SetTime
									rf.nextIndex[i].idx = nextIndex + 1
								}
								rf.nextIndex[i].mu.Unlock()

								// set matchIndex
								matchIndexUpdated := false
								rf.matchIndex[i].mu.Lock()
								rf.logMessage(fmt.Sprintf("rf.matchIndex[%d].setTime: %d, rf.matchIndex[%d].idx: %d", i, rf.matchIndex[i].setTime, i, rf.matchIndex[i].idx))
								if reply.SetTime > rf.matchIndex[i].setTime && rf.matchIndex[i].idx < nextIndex {
									rf.matchIndex[i].setTime = reply.SetTime
									rf.matchIndex[i].idx = nextIndex
									matchIndexUpdated = true
								}
								rf.matchIndex[i].mu.Unlock()
								if matchIndexUpdated {
									matchIndexCntMutex.Lock()
									matchIndexCnt[nextIndex]++
									cntValue := matchIndexCnt[nextIndex]
									matchIndexCntMutex.Unlock()
									rf.logMessage(fmt.Sprintf("setting matchIndexCnt[%d] + 1 to %d", nextIndex, cntValue + 1))

									rf.commitIndex.Lock()
									if nextIndex > rf.commitIndex.value && cntValue + 1 > len(rf.peers) / 2 && entries[0].Term == currentTerm {
										rf.commitIndex.value = nextIndex
										rf.logMessage(fmt.Sprintf("set rf.commitIndex to %d", nextIndex))
										rf.commitIndex.cond.Signal()
									}
									rf.commitIndex.Unlock()
								}
							}
						} else if !reply.TermError {
							rf.logMessage(fmt.Sprintf("append entry to server %d failed because of log inconsistency, reply.SetTime: %d, rf.nextIndex[i].setTime: %d, rf.nextIndex[i].idx: %d", i, reply.SetTime, rf.nextIndex[i].setTime, rf.nextIndex[i].idx))
							rf.nextIndex[i].mu.Lock()
							if reply.SetTime > rf.nextIndex[i].setTime && rf.nextIndex[i].idx > nextIndex - 1 {
								rf.nextIndex[i].setTime = reply.SetTime
								rf.nextIndex[i].idx = nextIndex - 1
							}
							rf.nextIndex[i].mu.Unlock()
						} else {
							rf.logMessage(fmt.Sprintf("append entry to server %d failed because of term error", i))
							rf.Lock()
							if reply.Term > rf.term {
								rf.term = reply.Term
								rf.voteFor = -1
								rf.status = 0
								rf.persist()
								rf.cond.Broadcast()
								rf.requestVoteTerm.Store(-1)
								rf.appendEntryTerm.Store(-1)
								rf.logMessage(fmt.Sprintf("during handling heartbeat response from server%d, setting term to %d", i, rf.term))
							}
							rf.Unlock()
						}
					} else {
						if entries == nil {
							rf.logMessage(fmt.Sprintf("send AppendEntry(heartbeat) to server %d failed, args: %v, time cost: %s", i, *args, elapsed))
						} else {
							rf.logMessage(fmt.Sprintf("send AppendEntry to server %d failed, args: %v, time cost: %s", i, *args, elapsed))
						}
					}

					if entries == nil {
						time.Sleep(time.Duration(120) * time.Millisecond)	// send heartbeat per 120ms
					}
				}
			}(i)
		}
	}	
}

func (rf *Raft) apply(applyCh chan ApplyMsg) {
	for !rf.killed() {
		rf.commitIndex.Lock()
		if rf.commitIndex.value > rf.lastApplied {
			rf.RLock()
			// test system expect index start from 1, so assign CommandIndex as rf.lastApplied + 2
			applyCh <- ApplyMsg{CommandValid: true, Command: rf.logs[rf.lastApplied + 1].Command, CommandIndex: rf.lastApplied + 2, 
			SnapshotValid: false, Snapshot: nil, SnapshotTerm: 0, SnapshotIndex: -1}
			rf.logMessage(fmt.Sprintf("apply index %d log entry", rf.lastApplied + 1))
			rf.RUnlock()
			rf.lastApplied++
		} else {
			rf.commitIndex.cond.Wait()
		}
		rf.commitIndex.Unlock()
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.initLogger()
	// Your initialization code here (2A, 2B, 2C).
	rf.term = 1
	rf.status = 0
	rf.voteFor = -1
	rf.commitIndex.value = -1
	rf.commitIndex.cond = sync.NewCond(&rf.commitIndex.Mutex)
	rf.lastApplied = -1
	rf.append_flag_cnt.Store(0)
	rf.cond = sync.NewCond(&rf.mu)
	rf.nextIndex = make([]settableIndex, len(rf.peers))
	rf.matchIndex = make([]settableIndex, len(rf.peers))
	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.apply(applyCh)

	return rf
}
